package com.heima.kafka;

import org.apache.kafka.clients.producer.*;
import org.junit.Test;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class ProducerQuickStart {
    @Test
    public void sendMsg() throws ExecutionException, InterruptedException {
        //编写配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"106.15.193.178:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        //核心对象的创建
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        //发送消息
        ProducerRecord<String, String> record = new ProducerRecord<>("topic","hello","kafka");
        RecordMetadata recordMetadata = producer.send(record).get();
        System.out.println(recordMetadata.offset());
        //关闭连接
        producer.close();
    }

    /*
    异步发送消息
     */
    @Test
    public void sendAsyncMsg() throws ExecutionException, InterruptedException {
        //编写配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"106.15.193.178:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.ACKS_CONFIG,"all");//确认机制
        properties.put(ProducerConfig.RETRIES_CONFIG,"10");//重试次数
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"gzip");//消息压缩

        //核心对象的创建
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        //发送消息
        ProducerRecord<String, String> record = new ProducerRecord<>("topic","hello","kafka");
       producer.send(record, new Callback() {
           @Override
           public void onCompletion(RecordMetadata recordMetadata, Exception e) {
               if (e != null){
                    e.printStackTrace();
               }
               System.out.println(recordMetadata);
           }
       });
        //关闭连接
        producer.close();
    }
}
